package com.kk.consumer;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

/**
 * Consumer ACK机制：
 * 1. 设置手动签收，rabbitmq.listener.simple.acknowledge-mode: manual
 * 2. 让监听器类实现ChannelAwareMessageListener接口
 * 3. 如果消息成功处理，则调用channel 的 basicAck() 签收
 * 4. 如果消息失败处理，则调用channel 的basicNack() 拒绝
 */
@Component
public class AckListener {


    /**
     * ttl
     * @param message
     */
    @RabbitListener(queues = "queue_ttl")
    public void receiveMessage2(Message message,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag)throws Exception{

        // 接收转换消息
        System.out.println (new String (message.getBody ( )));
        // 拒绝，重发（发送的时候设置 ttl 3s，所以拒绝后重复接收，知道3 s后停止）
        channel.basicNack (tag, true, true);
    }


    /**
     * Consumer 限流机制
     * 1. 确保ack机制为手动确认。
     * 2. listener-container配置属性
     *         rabbitmq.listener.simple.concurrency: 2,表示消费端每次从mq拉取 2条数据来消费，直到手动确认消费完毕后，才会继续拉去下一条消息
     * @param message
     */
    @RabbitListener(queues = "qos_queue")
    public void receiveMessage1(Message message,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag)throws Exception{

        Thread.sleep (1000);
        //1.获取消息
        System.out.println(new String(message.getBody()));

        //2. 处理业务逻辑

        //3. 签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);

    }

    /**
     * 消费者
     *
     * @param message 接收到的消息内容
     * @param channel 信道
     * @param tag 标记当前消息的tag，用于ACK
     */
    @RabbitListener(queues = "confirm_queue")
    public void receiveMessage(Message message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {

        try {
            //1、接收转换消息
            System.out.println (new String (message.getBody ( )));
            //2、处理业务逻辑
            System.out.println ("处理业务逻辑");
            //int i = 3 / 0;// 出现错误
            //3、手动签收
            channel.basicAck (tag, true);

        } catch (Exception e) {
//            e.printStackTrace ( );
            //4、拒绝签收

            /*
            第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
             */
            channel.basicNack (tag, true, true);
        }
    }
}
